package sc.analysis.rsi;



import java.io.IOException;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import sc.writable.PriceData;
import sc.writable.WritableArrayListFloat;



/**
 * Mapper of the Relative Strength Indicator analysis. Calculates the RSI from
 * price data.
 * 
 * TODO advanced RSI (RSI med/long term trend) see http://stockcharts.com/
 * school/doku.php?id=chart_school:technical_indicators:relative_strength_in
 * 
 * @author Quentin Rossé
 *
 */
public class RsiMapper extends Mapper<Text, PriceData, Text, FloatWritable>{
	
	/**
	 * the number of days for the indicator
	 */
	private static final int PERIOD = 14;
	
	/**
	 * data that will be written to the reducer
	 */
	private static FloatWritable fw = new FloatWritable();
	
	public void map(Text key, PriceData value, Context context) throws IOException, InterruptedException{
		WritableArrayListFloat closePrices = value.getClosePrices();
		int arraySize = closePrices.size();
		
		//for a n day RSI, we need n+1 price data
		if(arraySize > PERIOD){
			float lastPrice = closePrices.getFloat(arraySize-1);
			float newPrice;
			float delta;
			float sumGain = 0f, sumLoss = 0f;
			
			for(int i=arraySize-2; i>=arraySize- PERIOD - 1; i--){
				newPrice = closePrices.getFloat(i);
				delta = newPrice - lastPrice;
				if(delta > 0)
					sumGain += delta;
				else
					sumLoss -= delta;
				lastPrice = newPrice;
			}
			if(sumLoss == 0)
				fw.set(100f);
			else
				fw.set(100f-100f/(1f+sumGain/sumLoss));
			context.write(key, fw);
		}
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
}
